#ifndef _M_QUEUEMESSAGE_H_
#define _M_QUEUEMESSAGE_H_

#include <iostream>
#include <google/protobuf/map.h>
#include <string>
#include <vector>
#include <cassert>
#include <sstream>
#include <functional>
#include <mutex>
#include <unordered_map>
#include <string>
#include <list>

#include "../mqcommon/mqProto.pb.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helper.hpp"

namespace MyRabbitMQ
{
    #define DATAFILE_SUBFIX ".mqd"
    #define TMPFILE_SUBFIX ".mqd.tmp"

    // proto消息指针
    using pbMessagePtr = std::shared_ptr<protoMQ::Message>;

    // 队列消息数据持久化管理类
    // 我们在基础路径下, 每个队列都创建一个文件, 单独的管理持久化消息
    class MessageMapper
    {
    public:
        using ptr = std::shared_ptr<MessageMapper>;
        MessageMapper(std::string &basedir, const std::string &qname)
            : m_qname(qname)
        {
            // 如果路径非目录路径 ,则添加/
            if (basedir.back() != '/')
                basedir.push_back('/');

            // 如果基本目录路径不存在则创建, 创建失败则直接退出
            if (!FileHelper(basedir).exists())
                assert(FileHelper::createDirectory(basedir));

            // 实例化成员变量
            m_dbfilename = basedir + qname + DATAFILE_SUBFIX;
            m_dbfilename_TMP = basedir + qname + TMPFILE_SUBFIX;

            // 创建数据持久化文件
            assert(createMsgFile());
        }

        // 创建数据持久化文件
        bool createMsgFile()
        {
            // 如果已经存在则不用创建
            if (FileHelper(m_dbfilename).exists())
                return true;

            // 不存在则创建
            return FileHelper::createFile(m_dbfilename);
        }

        // 删除数据持久化文件
        void removeMsgFile()
        {
            FileHelper::removeFile(m_dbfilename);
            FileHelper::removeFile(m_dbfilename_TMP);
        }

        // 向指定文件新增消息数据
        bool insert(const pbMessagePtr& ptr, const std::string &filename)
        {
            // 获取有效载荷
            std::string message = ptr->payload().SerializeAsString();

            // 获取文件长度
            FileHelper fileHelper(filename);
            size_t fsize = fileHelper.size();

            // 写入文件数据, 先写入数据长度的数据
            size_t msgsize = message.size();

            bool ret = fileHelper.write((char *)&msgsize, fsize, sizeof(size_t));
            if (!ret)
            {
                return false;
            }

            ret = fileHelper.write(&message[0], fsize + sizeof(size_t), msgsize);
            if (!ret)
            {
                return false;
            }

            // 写入成功, 设置消息数据的偏移量和长度
            ptr->set_offset(fsize + sizeof(size_t));
            ptr->set_length(msgsize);

            return true;
        }

        // 从指定文件删除消息数据
        bool remove(const pbMessagePtr &ptr, const std::string &filename)
        {
            // 将消息的有效标记位设置为0
            ptr->mutable_payload()->set_isvaild("0");

            // 获取序列化后的, 标记位为0的数据
            std::string message = ptr->payload().SerializeAsString();

            // 获取数据的长度
            size_t msgs = message.size();

            // 如果数据长度不一致则直接报错
            if (msgs != ptr->length())
            {
                return false;
            }

            FileHelper fileHelper(filename);
            bool ret = fileHelper.write(&message[0], ptr->offset(), msgs);

            if (!ret)
            {
                return false;
            }

            return true;
        }

        // 向数据文件新增(选择是源文件还是临时文件)
        bool insert(const pbMessagePtr &ptr)
        {
            return this->insert(ptr, m_dbfilename);
        }

        // 从数据文件删除(同上)
        bool remove(const pbMessagePtr &ptr)
        {
            return this->remove(ptr, m_dbfilename);
        }

        // 从文件中加载有效数据
        bool load(std::list<pbMessagePtr> &result)
        {
            // 定义偏移量, 返回值, 获取文件总大小
            size_t offset = 0;
            FileHelper fileHelper(m_dbfilename);
            size_t fsize = fileHelper.size();

            while (offset < fsize)
            {
                size_t len;

                // 读取消息长度
                if (!fileHelper.read((char *)&len, offset, sizeof(size_t)))
                {
                    return false;
                }

                offset += sizeof(size_t);

                //重设data大小
                std::string data(len,'\0');

                // 读取消息内容
                if (!fileHelper.read(&data[0], offset, len))
                {
                    return false;
                }

                offset += len;

                pbMessagePtr ptr = std::make_shared<protoMQ::Message>();

                // 检测数据的有效位情况
                ptr->mutable_payload()->ParseFromString(data);

                // 如果消息是无效消息,则直接略过
                if (ptr->payload().isvaild() == "0")
                    continue;

                // 如果消息有效, 保存至链表中
                result.push_back(ptr);
            }
            return true;
        }

        // 垃圾回收函数,清理文件并重新构造, 返回文件有效数据(持久化层面垃圾回收)
        std::list<pbMessagePtr> gc()
        {
            // 定义返回值
            std::list<pbMessagePtr> res;

            // 加载文件所有有效数据
            if (!this->load(res))
            {
                LOG(logLevel::ERROR) << "执行失败:加载有效数据失败" << std::endl;
                return res;
            }

            // 创建临时文件
            if (!FileHelper::createFile(m_dbfilename_TMP))
            {
                LOG(logLevel::ERROR) << "执行失败:创建临时文件失败" << std::endl;
                return res;
            }

            // 将有效数据序列化存储到临时文件中
            for (auto &f : res)
            {
                if (!this->insert(f, m_dbfilename_TMP))
                {
                    LOG(logLevel::ERROR) << "执行失败:有效数据添加至临时文件时失败" << std::endl;
                    return res;
                }
            }

            // 删除源文件, 更改临时文件文件名
            if (!FileHelper::removeFile(m_dbfilename))
            {
                LOG(logLevel::ERROR) << "执行失败:删除源数据文件失败" << std::endl;
                return res;
            }

            FileHelper fileHelper(m_dbfilename_TMP);

            if (!fileHelper.rename(m_dbfilename))
            {
                LOG(logLevel::ERROR) << "执行失败:更改临时文件文件名失败" << std::endl;
                return res;
            }

            return res;
        }

    private:
        std::string m_qname;
        std::string m_dbfilename;
        std::string m_dbfilename_TMP;
    };

    // 队列消息数据管理类--以队列为单位
    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

        QueueMessage(const std::string &qname, std::string &basedir)
            : m_qname(qname),
              m_mapper(basedir, qname), m_total_durable_count(0), m_valid_durable_count(0)
        {
        }

        // 恢复历史消息
        void recovery()
        {
            // 加锁
            std::unique_lock<std::mutex>(m_mtx);

            //清理历史数据
            m_durable_msgs.clear();

            // 从持久化管理句柄中恢复所有等待推送的历史消息
            m_msgs = m_mapper.gc();

            // 这些消息都是持久化消息, 所以需要推送到持久化消息hash中
            for (auto &f : m_msgs)
            {
                m_durable_msgs.insert(std::make_pair(f->payload().properties().id(), f));
            }

            // 设置持久化消息数量
            m_total_durable_count = m_valid_durable_count = m_msgs.size();
        }

        // 添加消息
        bool insert(const protoMQ::BaseProperties *base, const std::string &data,
                    bool queue_isDurable) // 这里需要队列的持久化信息, 因为单纯消息持久化是无意义的, 必须队列和消息都持久化才能存储到文件中
        {
            // 构造消息对象
            auto ptr = std::make_shared<protoMQ::Message>();

            // 设置消息对象属性
            ptr->mutable_payload()->set_data(data);
            // 先设置消息默认在持久化数据中无效, 有效的后边会改
            ptr->mutable_payload()->set_isvaild("0");

            if (base != nullptr)
            {
                protoMQ::DeliveryMode mode = queue_isDurable == true ? base->mode() : protoMQ::DeliveryMode::undurable;
                ptr->mutable_payload()->mutable_properties()->set_id(base->id());
                ptr->mutable_payload()->mutable_properties()->set_routing_key(base->routing_key());
                ptr->mutable_payload()->mutable_properties()->set_mode(mode);
            }
            else
            {
                protoMQ::DeliveryMode mode = queue_isDurable == true ? protoMQ::DeliveryMode::durable : protoMQ::DeliveryMode::undurable;
                ptr->mutable_payload()->mutable_properties()->set_id(UuidHelper::uuid());
                ptr->mutable_payload()->mutable_properties()->set_routing_key("");
                ptr->mutable_payload()->mutable_properties()->set_mode(mode);
            }

            //联合测试
            //LOG(logLevel::normal) << "联合测试: 消息的id为" << ptr->mutable_payload()->properties().id() << std::endl;

            // 加锁
            std::unique_lock<std::mutex>(m_mtx);

            // 如果需要持久化, 那么还需要操作持久化管理句柄
            if (ptr->payload().properties().mode() == protoMQ::DeliveryMode::durable)
            {
                ptr->mutable_payload()->set_isvaild("1");
                
                if (!m_mapper.insert(ptr))
                {
                    LOG(logLevel::ERROR) << "执行失败:进行持久化时出现问题" << std::endl;
                    return false;
                }


                m_durable_msgs.insert(std::make_pair(ptr->payload().properties().id(), ptr));

                ++m_total_durable_count, ++m_valid_durable_count;
            }

            // 向待推送消息列表添加数据
            m_msgs.push_back(ptr);

            return true;
        }
        // 从待确认hash中删除消息
        bool remove(const std::string &id)
        {
            // 加锁
            std::unique_lock<std::mutex>(m_mtx);

            // 在待确认消息hash中寻找对应id的消息,找不到则返回(只有待确认的消息才能允许被直接删除)
            auto it = m_waitack_msgs.find(id);
            if (it == m_waitack_msgs.end())
            {
                return true;
            }

            // 找到消息, 准备删除
            // 先判断消息是否持久化
            if (it->second->payload().isvaild() == "1")
            {
                // 从mapper中删除对应消息
                if (!m_mapper.remove(it->second))
                {
                    LOG(logLevel::ERROR) << "执行失败:进行持久化时出现问题" << std::endl;
                    return false;
                }

                // 从持久化消息hash中删除消息
                m_durable_msgs.erase(id);

                // 删除成功, 计数减少
                --m_valid_durable_count;

                // 判断是否进行垃圾回收, 需要则进行
                if (this->isNeedGC())
                {
                    this->gc();
                }
            }

            // 内存层面删除
            m_waitack_msgs.erase(it);
            return true;
        }

        // 获取队首消息
        pbMessagePtr front()
        {
            std::unique_lock<std::mutex>(m_mtx);
            // 如果待推送消息列表无消息则返回
            if (m_msgs.size() == 0)
            {
                return pbMessagePtr();
            }

            auto res = m_msgs.front();
            m_msgs.pop_front();
            m_waitack_msgs.insert(std::make_pair(res->payload().properties().id(), res));
            return res;
        }

        // 清除队列消息数据
        void clear()
        {
            std::unique_lock<std::mutex>(m_mtx);
            m_mapper.removeMsgFile();
            m_msgs.clear();
            m_durable_msgs.clear();
            m_waitack_msgs.clear();
            m_total_durable_count = m_valid_durable_count = 0;
        }

        // 获取总持久化消息数量
        size_t getTotalDurableSize()
        {
            return m_total_durable_count;
        }

        // 获取有效持久化消息数量
        size_t getValidDurableSize()
        {
            return m_valid_durable_count;
        }

        // 获取待推送消息数量
        size_t getToBePushedSize()
        {
            return m_msgs.size();
        }

        // 获取待确认消息数量
        size_t getWaitackSize()
        {
            return m_waitack_msgs.size();
        }

    private:
        std::string m_qname;                                        // 队列名称
        MessageMapper m_mapper;                                     // 持久化管理句柄
        std::list<pbMessagePtr> m_msgs;                               // 等待推送的消息
        std::unordered_multimap<std::string, pbMessagePtr> m_durable_msgs; // 持久化消息,使用id寻找
        std::unordered_multimap<std::string, pbMessagePtr> m_waitack_msgs; // 待确认消息,使用id寻找
        size_t m_total_durable_count;                               // 总持久化消息数量
        size_t m_valid_durable_count;                               // 有效持久化消息数据
        std::mutex m_mtx;

        // 判断是否需要进行垃圾回收(单独使用非锁安全)
        bool isNeedGC()
        {
            return m_total_durable_count >= 2000 && (m_valid_durable_count * 10 / m_total_durable_count) < 5;
        }

        // 垃圾回收(内存层面垃圾回收)(单独使用非锁安全)
        void gc()
        {
            // 持久化层面垃圾回收, 获取所有有效消息
            auto msgs = m_mapper.gc();

            for (auto &ptr : msgs)
            {
                // 查找id对应的消息
                auto it = m_durable_msgs.find(ptr->payload().properties().id());

                // 如果未找到, 那么是有误的, 因为明明持久化了, 但消息缺未被管理
                if (it == m_durable_msgs.end())
                {
                    LOG(logLevel::WARNING) << "有一条持久化消息未被内存管理! 即将添加..." << std::endl;
                    m_durable_msgs.insert(std::make_pair(ptr->payload().properties().id(), ptr));
                    m_msgs.push_back(ptr);
                    continue;
                }

                // 如果找到,则对内存中已经管理的消息进行修改
                it->second->set_length(ptr->length());
                it->second->set_offset(ptr->offset());
            }

            // 更新数量
            m_total_durable_count = m_valid_durable_count = m_durable_msgs.size();
        }
    };
    // 队列消息数据内存层面管理类(整体管理)
    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;

        MessageManager(const std::string &basedir)
            : m_basedir(basedir)
        {
        }
        // 声明队列消息管理
        bool declareQueueMessageManager(const std::string &qname)
        {
            QueueMessage::ptr ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);
                // 查找是否已经存在, 如果已经存在则无需声明
                if (m_queuemsgs.find(qname) != m_queuemsgs.end())
                    return true;

                // 不存在则添加
                ptr = std::make_shared<QueueMessage>(qname, m_basedir);
                m_queuemsgs.insert(std::make_pair(qname, ptr));
            }
            ptr->recovery();
            return true;
        }

        // 删除队列消息管理
        bool deleteQueueMessageManager(const std::string &qname)
        {
            QueueMessage::ptr ptr = std::make_shared<QueueMessage>(qname, m_basedir);
            {
                std::unique_lock<std::mutex>(m_mtx);
                // 查找是否已经存在, 不存在则直接返回
                auto it = m_queuemsgs.find(qname);

                if (it == m_queuemsgs.end()) return true;

                //存在则删除
                ptr = it->second;
                m_queuemsgs.erase(it);
            }

            ptr->clear();
            return true;
        }
        // 向指定队列消息管理增加消息
        bool insert(const std::string &qname, const protoMQ::BaseProperties *base,
                    const std::string &data, bool queue_isDurable)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);
                auto it = m_queuemsgs.find(qname);
                if (it == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = it->second;
            }
            return ptr->insert(base, data, queue_isDurable);
        }

        // 获取指定队列消息管理队首消息
        pbMessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);
                auto it = m_queuemsgs.find(qname);
                if (it == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return pbMessagePtr();
                }
                ptr = it->second;
            }

            return ptr->front();
        }
        // 确认指定队列消息
        bool ack(const std::string &qname, const std::string &id)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);

                auto qit = m_queuemsgs.find(qname);
                if (qit == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = qit->second;
            }
            return ptr->remove(id);
        }

        // 获取指定队列消息数量--总持久化数量
        size_t totalSize(const std::string& qname)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);

                auto qit = m_queuemsgs.find(qname);
                if (qit == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = qit->second;
            }

            return ptr->getTotalDurableSize();           
        }

        // 获取指定队列消息数量--有效持久化数量
        size_t validSize(const std::string& qname)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);

                auto qit = m_queuemsgs.find(qname);
                if (qit == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = qit->second;
            }

            return ptr->getValidDurableSize();          
        }

        // 获取指定队列消息数量--待推送消息数量
        size_t toBePushedSize(const std::string& qname)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);

                auto qit = m_queuemsgs.find(qname);
                if (qit == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = qit->second;
            }

            return ptr->getToBePushedSize();           
        }

        // 获取指定队列消息数量--待确认消息数量
        size_t waitackSize(const std::string& qname)
        {
            std::shared_ptr<QueueMessage> ptr;
            {
                std::unique_lock<std::mutex>(m_mtx);

                auto qit = m_queuemsgs.find(qname);
                if (qit == m_queuemsgs.end())
                {
                    LOG(logLevel::ERROR) << "执行失败:指定队列不存在!" << std::endl;
                    return false;
                }

                ptr = qit->second;
            }

            return ptr->getWaitackSize();           
        }

        // 清理所有队列的所有消息
        void clear()
        {
            std::unique_lock<std::mutex>(m_mtx);

            for(auto & msg : m_queuemsgs)
            {
                msg.second->clear();
            }

            m_queuemsgs.clear();
        }
    private:
        std::mutex m_mtx;
        std::string m_basedir;
        std::unordered_map<std::string, QueueMessage::ptr> m_queuemsgs;
    };
};

#endif